package com.shiguiwu.springboot3.kafka.listener;

import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Optional;

/**
 * @description: 监听消费消息 测试两个客户端
 * @author: stone
 * @date: Created by 2023/11/13 16:38
 * @version: 1.0.0
 * @pakeage: com.shiguiwu.springboot3.kafka.listener
 */
@Component
@Slf4j
public class KafkaConsumer2Listener {


    //监听消费
    @KafkaListener(topics = {"test520","test008"})
    public void onNormalMessage1(ConsumerRecord<String, Object> record, Acknowledgment ack) {
        try {
            log.warn("第二个客户端topic = {},record.partition={},value={} ", record.topic(), record.partition(), record.value());
        } finally {
            Optional.ofNullable(ack).ifPresent(Acknowledgment::acknowledge);
        }

    }

    //监听消费
    @KafkaListener(topics = {"shigwTopic"})
    public void multiPartitions(ConsumerRecord<String, Object> record, Acknowledgment ack) {
        try {
            log.warn("第二个客户端topic = {},record.partition={},value={} ", record.topic(), record.partition(), record.value());
        } finally {
            Optional.ofNullable(ack).ifPresent(Acknowledgment::acknowledge);
        }

    }



}
